package com.xinxuan.contentcenter.rocketmq;

import java.util.Objects;

import javax.annotation.Resource;

import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSON;
import com.xinxuan.contentcenter.domain.RocketmqTransactionLog;
import com.xinxuan.contentcenter.dto.ShareAuditDTO;
import com.xinxuan.contentcenter.mapper.RocketmqTransactionLogMapper;
import com.xinxuan.contentcenter.service.ShareService;

import lombok.extern.slf4j.Slf4j;

/**
 * @author xinsl
 * @date 2021/11/25
 */
@Slf4j
@Service
@RocketMQTransactionListener(txProducerGroup = "tx-add-bonus-group")
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {

    private final ShareService shareService;

    @Resource
    private RocketmqTransactionLogMapper rocketmqTransactionLogMapper;

    public TransactionListenerImpl(ShareService shareService) {
        this.shareService = shareService;
    }

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        MessageHeaders headers = message.getHeaders();
        String transactionId = (String)headers.get(RocketMQHeaders.TRANSACTION_ID);
        String shardId = (String)headers.get("shard_id");
        String dtoStr = (String)headers.get("dto");
        ShareAuditDTO shareAuditDTO = JSON.parseObject(dtoStr, ShareAuditDTO.class);
        try {
            shareService.auditByIdWiithRocketMqLog(Integer.parseInt(shardId), shareAuditDTO, transactionId);
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            log.info("信息提交失败，失败原因：{}", e);
            return RocketMQLocalTransactionState.ROLLBACK;
        }

    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        MessageHeaders headers = message.getHeaders();
        String transactionId = (String)headers.get(RocketMQHeaders.TRANSACTION_ID);
        RocketmqTransactionLog transactionLog = rocketmqTransactionLogMapper
            .selectOne(RocketmqTransactionLog.builder().transactionId(transactionId).build());
        if (Objects.equals(null, transactionLog)) {
            return RocketMQLocalTransactionState.COMMIT;
        }
        return RocketMQLocalTransactionState.ROLLBACK;
    }
    // @Override
    // public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
    // String transactionId = msg.getTransactionId();
    // String shardId = msg.getProperty("shard_id");
    // try {
    // ShareAuditDTO shareAuditDTO = arg instanceof ShareAuditDTO ? ((ShareAuditDTO)arg) : null;
    // shareService.auditByIdWiithRocketMqLog(Integer.parseInt(shardId), shareAuditDTO, transactionId);
    // return LocalTransactionState.COMMIT_MESSAGE;
    // } catch (Exception e) {
    // log.info("信息提交失败，失败原因：{}", e);
    // return LocalTransactionState.ROLLBACK_MESSAGE;
    // }
    // }

    // @Override
    // public LocalTransactionState checkLocalTransaction(MessageExt msg) {
    // String transactionId = msg.getTransactionId();
    // RocketmqTransactionLog transactionLog = rocketmqTransactionLogMapper
    // .selectOne(RocketmqTransactionLog.builder().transactionId(transactionId).build());
    // if (Objects.equals(null, transactionLog)) {
    // return LocalTransactionState.COMMIT_MESSAGE;
    // }
    // return LocalTransactionState.ROLLBACK_MESSAGE;
    // }

}
